﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace FrameWork.Core.ThreadPool
{
    public class TaskPools
    {
        #region ___单列___
        private static TaskPools instance = null;
        private static object obj = new object();
        public static TaskPools Instance
        {
            get
            {
                if (instance == null)
                {
                    lock (obj)
                    {
                        if (instance == null)
                        {
                            instance = new TaskPools();
                        }
                    }
                }
                return instance;
            }
        }
        #endregion

        #region ___字段___
        /// <summary>
        /// 待处理数据队列
        /// </summary>
        private ConcurrentQueue<TaskQueueItem> queue = new ConcurrentQueue<TaskQueueItem>();
        /// <summary>
        /// 运行中的线程
        /// </summary>
        private ConcurrentDictionary<int, CancellationTokenSource> taskThreads = new ConcurrentDictionary<int, CancellationTokenSource>();
        /// <summary>
        /// 数据锁
        /// </summary>
        private object countLock = new object();//计数锁
        /// <summary>
        /// 完成任务数
        /// </summary>
        private int completedCount = 0;//任务完成次数
        /// <summary>
        /// 数据锁
        /// </summary>
        public AutoResetEvent ConsumerEvent = new AutoResetEvent(false);
        /// <summary>
        /// 消费者线程
        /// </summary>
        private Thread threadConsumer;
        /// <summary>
        /// 线程编号
        /// </summary>
        private int Number = 0;

        private bool Running = true;
        #endregion

        #region ___构造函数___
        public TaskPools()
        { }
        public TaskPools(int maxThreadCount)
        {
            this.MaxThreadCount = maxThreadCount;
        }
        #endregion

        #region ___属性___
        /// <summary>
        /// 任务分组号
        /// </summary>
        public int GroupId { get; set; }
        /// <summary>
        /// 最大线程数
        /// </summary>
        public int MaxThreadCount { get; set; } = 10000;
        /// <summary>
        /// 最小线程数
        /// </summary>
        public int MinThreadCount { get; set; } = 0;
        /// <summary>
        /// 当前线程数
        /// </summary>
        public int ThreadCount
        {
            get
            {
                return this.taskThreads.Count;
            }
        }
        /// <summary>
        /// 当前未完成的任务数
        /// </summary>
        public int TaskCount
        {
            get
            {
                return this.queue.Count;
            }
        }
        /// <summary>
        /// 当前完成的任务数
        /// </summary>
        public int CompletedCount
        {
            get
            {
                lock (countLock)
                {
                    return this.completedCount;
                }
            }
        }
        #endregion

        #region ___事件___
        /// <summary>
        /// 线程数/未完成任务数/完成任务数 变化事件
        /// </summary>
        public event EventHandler<CountChangedArgs> CountChanged;
        /// <summary>
        /// 触发CountChanged事件
        /// </summary>
        private void OnCountChanged()
        {
            try
            {
                Console.WriteLine("组【" + GroupId + "】,已执行【" + CompletedCount + "】个,待执行【" + TaskCount + "】个");
                CountChanged?.Invoke(this, new CountChangedArgs(this.TaskCount, this.ThreadCount, this.CompletedCount));
            }
            catch { }
        }
        #endregion

        #region ___生产者___
        /// <summary>
        /// 向任务池中添加一个任务
        /// </summary>
        /// <param name="action"></param>
        public void Enqueue(Action action, CancellationTokenSource cts = null)
        {
            Enqueue(++Number, action, cts, false, 0);
        }
        /// <summary>
        /// 向任务池中添加一个任务
        /// </summary>
        /// <param name="action">任务执行方法</param>
        /// <param name="reExecute">是否重复执行任务</param>
        /// <param name="reCount">重复执行次数，0表示无限制</param>
        public void Enqueue(Action action, CancellationTokenSource cts, bool reExecute, int reCount)
        {
            Enqueue(++Number, action, cts, reExecute, reCount);
        }
        /// <summary>
        /// 向任务池中添加一个任务
        /// </summary>
        /// <param name="id">任务id</param>
        /// <param name="name">任务名称</param>
        /// <param name="action">任务执行方法</param>
        /// <param name="reExecute">是否重复执行任务</param>
        /// <param name="reCount">重复执行次数，小于等于0表示无限制</param>
        public void Enqueue(int id, Action action, CancellationTokenSource cts, bool reExecute, int reCount)
        {
            if (action == null) return;
            queue.Enqueue(new TaskQueueItem(action) { GroupID = GroupId, ID = id, Cts = cts, ReExecute = reExecute, ReCount = reCount, ExecCount = 0 });
            Console.WriteLine(
                 "组【" + GroupId + "】" +
                 "添加任务【" + id + "】成功," +
                 "重复执行【" + (reExecute ? "是" : "否") + "】," +
                 (reExecute ? "重复【" + (reCount == 0 ? "无数" : reCount.ToString()) + "】次" : string.Empty)
                );
            if (null == threadConsumer)
            {
                StartConsumer();
                Console.WriteLine("组【" + GroupId + "】启动任务池成功");
            }
        }

        /// <summary>
        /// 阻塞等待多任务完成
        /// </summary>
        /// <param name="func"></param>
        public void Enqueue(IEnumerable<Action> actions)
        {
            var tasklst = new List<Task>();
            actions.ToList().ForEach((t) =>
            {
                tasklst.Add(Task.Factory.StartNew(t));
            });
            Task.WaitAll();
        }

        /// <summary>
        /// 阻塞等待多任务完成
        /// </summary>
        /// <param name="func"></param>
        public void Enqueue(IEnumerable<Action> actions, Action notice)
        {
#if NET40
#else
            var blockQueue = new BlockingCollection<TaskQueueItem>();
            /////生产任务////
            actions.ToList().ForEach((t) =>
            {
                blockQueue.Add(new TaskQueueItem(t) { GroupID = GroupId, ID = ++Number });
            });
            blockQueue.CompleteAdding(); //标记生产完成
            /////消费任务//////
            Task.WaitAll(Task.Run(() => { BlockTaskAction(blockQueue); }));
            notice?.Invoke();
#endif
        }
        /// <summary>
        /// 先执行A，A执行成功后执行B
        /// </summary>
        /// <param name="func"></param>
        public void Enqueue(Action condition, Action after)
        {
#if NET40
#else
            Task.Run(() => { condition.Invoke(); }).ContinueWith((r) =>
            {
                if (r.IsFaulted)
                {
                    Console.WriteLine(r.Exception.GetBaseException());
                }
                else
                    after.Invoke();
            });
#endif
        }
        #endregion

        #region ___消费者___
        /// <summary>
        /// 启动消费者线程
        /// </summary>
        public void StartConsumer()
        {
            ThreadHelper.StartBgThread(ref threadConsumer, new ParameterizedThreadStart(Consumer), null, true);
        }
        /// <summary>
        /// 停止消费者
        /// </summary>
        public void StopConsumer()
        {
            Running = false;
            this.taskThreads.Clear();
            threadConsumer.Join();
        }
        /// <summary>
        /// 处理队列中的数据
        /// </summary>
        /// <param name="state"></param>
        public void Consumer(object state)
        {
            while (Running)
            {
                try
                {
                    ConsumerEvent.WaitOne(500);
                    if (queue.Count > 0)
                    {
                        ///////////执行的线程数量小于最大线程数量/////////////
                        if (this.taskThreads.Count < this.MaxThreadCount)
                        {
                            if (queue.TryDequeue(out TaskQueueItem task))
                            {
                                if (null != task)
                                {
                                    if (task.Cts == null)
                                    {
                                        Task.Factory.StartNew(() => this.TaskAction(task));
                                        this.taskThreads.AddOrUpdate(task.ID, task.Cts, (k, v) => task.Cts);
                                    }
                                    else
                                    {
                                        if (!task.Cts.IsCancellationRequested)
                                        {
                                            Task.Factory.StartNew(() => this.TaskAction(task), task.Cts.Token);
                                            this.taskThreads.AddOrUpdate(task.ID, task.Cts, (k, v) => task.Cts);
                                        }
                                    }
                                }
                            }
                        }
                    }
                    ConsumerEvent.Reset();
                }
                catch (Exception ex)
                {
                    ConsumerEvent.Reset();
                    throw new Exception("组【" + GroupId + "】消费者线程执行出错", ex);
                }
            }
        }
        /// <summary>
        /// 执行任务
        /// </summary>
        public void TaskAction(object state)
        {
            if (null != state)
            {
                var task = state as TaskQueueItem;
                if (null != task.Cts && task.Cts.IsCancellationRequested)
                    return;

                try
                {
                    Console.WriteLine("开始执行组【" + GroupId + "】任务【" + task.ID + "】");
                    task.Action.Invoke();
                }
                catch (Exception ex)
                {
                    throw new Exception("组【" + GroupId + "】任务【" + task.ID + "】执行出错", ex);
                }
                task.ExecCount += 1;

                //////////打印执行结果日志/////////
                Console.WriteLine(
                    "组【" + task.GroupID + "】" +
                    "任务【" + task.ID + "】" +
                    "第【" + task.ExecCount + "】次执行完成" +
                    (task.ReExecute ? "重复【" + (task.ReCount == 0 ? "无数" : task.ReCount.ToString()) + "】次" : string.Empty)
                    );

                if ((task.ReExecute && task.ReCount == 0) || (task.ReExecute && task.ReCount > 0 && task.ExecCount < task.ReCount))
                {
                    //////满足重复执行条件重新加入队列待执行//////
                    queue.Enqueue(task);
                }
                lock (countLock)
                {
                    this.completedCount++;
                }
                //////////移除执行完的任务/////////
                if (TryRemoveThread(task))
                {
                    Console.WriteLine("组【" + GroupId + "】移除任务【" + task.ID + "】成功");
                }

                OnCountChanged();

                Thread.Sleep(50);

                //if (this.taskThreads.Count == 0)
                //{
                //    Console.WriteLine("停止线程池");
                //    StopConsumer(); 
                //}

            }
        }
        /// <summary>
        /// 执行并退出
        /// </summary>
        public void BlockTaskAction(BlockingCollection<TaskQueueItem> blockQueue)
        {
            Thread.Sleep(100);
            while (!blockQueue.IsCompleted)
            {
                var x = blockQueue.Take();
                x?.Action();
                Thread.Sleep(100);
            }
        }
        #endregion

        #region ___私有方法___
        /// <summary>
        /// 如果条件允许，减少一个线程
        /// </summary>
        /// <returns></returns>
        private bool TryRemoveThread(TaskQueueItem task)
        {
            if (this.taskThreads.TryRemove(task.ID, out CancellationTokenSource thread))
            {
                //if (null != thread)
                //    thread.Cancel();
                return true;
            }
            return false;
        }
        /// <summary>
        /// 清空数据
        /// </summary>
        public void Clear()
        {
            this.taskThreads.Clear();
            while (this.queue.TryDequeue(out TaskQueueItem item)) { }
            OnCountChanged();
        }
        #endregion
    }
}
